package drds.global_transaction.resource_manager.proxy.resource_manager;

import drds.common.$;
import drds.common.Constants;
import drds.global_transaction.api.api.BranchTransactionStatus;
import drds.global_transaction.api.exception.GlobalTransactionException;
import drds.global_transaction.api.exception.TransactionExceptionCode;
import drds.global_transaction.resource_manager.ConnectionContext;
import drds.global_transaction.resource_manager.ResourceManager;
import drds.global_transaction.resource_manager.execute.LockConflictException;
import drds.global_transaction.resource_manager.model.Column;
import drds.global_transaction.resource_manager.model.Image;
import drds.global_transaction.resource_manager.proxy.datasource.AbstractConnectionProxy;
import drds.global_transaction.resource_manager.sql_recognizer.SqlType;
import drds.global_transaction.resource_manager.undolog.UndoLog;
import drds.global_transaction.resource_manager.undolog.UndoLogManager;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.sql.Connection;
import java.sql.SQLException;

@Slf4j
public class ConnectionProxy extends AbstractConnectionProxy
{

	@Setter
	@Getter
	private ConnectionContext connectionContext = new ConnectionContext();

	public ConnectionProxy(DataSourceProxy dataSourceProxy, Connection connection)
	{
		super(dataSourceProxy, connection);
	}

	public void bind(String globalTransactionIdWithAddressAndPort)
	{
		connectionContext.bind(globalTransactionIdWithAddressAndPort);
	}

	private void branchTransactionRegister() throws GlobalTransactionException
	{
		Long branchTransactionId = ResourceManager.get().branchTransactionRegister(//
				getDataSourceProxy().getResourceId(), //
				null, //
				connectionContext.getGlobalTransactionIdWithAddressAndPort(), //
				connectionContext.buildLockRecordId());//
		connectionContext.setBranchTransactionId(branchTransactionId);
	}

	/**
	 * @param image 必须包含主键
	 * @return tableNameAndLockRecordIdsStringsString的一部分,
	 *         所以事务协调器方为tableNameAndLockRecordIdsStringsString,
	 *         但是资源管理器为lockRecordIdsString
	 */
	private String buildLockRecordIdsString(Image image)
	{
		if (image.size() == 0)
		{
			return null;
		}
		StringBuilder sb = new StringBuilder();
		sb.append(image.getTableMetaData().getTableName());
		sb.append(":");
		boolean flag = false;
		for (Column column : image.getPrimaryKeyColumnList())
		{
			if (flag)
			{
				sb.append(",");
			} else
			{
				flag = true;
			}
			sb.append(column.getValue());
		}
		return sb.toString();
	}

	private UndoLog buildUndoLog(SqlType sqlType, String tableName, Image beforeImage, Image afterImage)
	{
		UndoLog undoLog = new UndoLog();
		undoLog.setSqlType(sqlType);
		undoLog.setTableName(tableName);
		undoLog.setBeforeImage(beforeImage);
		undoLog.setAfterImage(afterImage);
		return undoLog;
	}
	//

	/**
	 * 多次提交,多次创建分支事务
	 */
	@Override
	public void commit() throws SQLException
	{
		if (connectionContext.inGlobalTransaction())
		{
			try
			{
				branchTransactionRegister();// 提交的时候才进行分支事务的创建
			} catch (GlobalTransactionException e)
			{
				recognizeLockKeyConflictException(e);
			}

			try
			{
				if (connectionContext.hasUndoLog())
				{
					UndoLogManager.flushUndoLogs(this);
				}
				connection.commit();
			} catch (Throwable ex)
			{
				report(false);
				if (ex instanceof SQLException)
				{
					throw (SQLException) ex;
				} else
				{
					throw new SQLException(ex);
				}

			}
			report(true);
			connectionContext.reset();
		} else
		{
			connection.commit();
		}
	}

	/**
	 * 判断是否已经lock
	 */
	public void lockRecordIdsCheck(Image image) throws SQLException
	{
		String lockRecordIdsString = buildLockRecordIdsString(image);
		try
		{
			boolean lock = ResourceManager.get().lockRecordIdsCheck(getDataSourceProxy().getResourceId(), connectionContext.getGlobalTransactionIdWithAddressAndPort(), lockRecordIdsString);
			if (!lock)
			{
				throw new LockConflictException();
			}
		} catch (GlobalTransactionException e)
		{
			recognizeLockKeyConflictException(e);
		}
	}

	//
	public void prepareUndoLog(SqlType sqlType, String tableName, Image beforeImage, Image afterImage) throws SQLException
	{
		Image lockKeyRecordIdsImage = afterImage;
		if (sqlType == SqlType.delete)
		{
			lockKeyRecordIdsImage = beforeImage;
		}
		String lockRecordIdsString = buildLockRecordIdsString(lockKeyRecordIdsImage);
		connectionContext.addLockRecordIdsString(lockRecordIdsString);
		UndoLog undoLog = buildUndoLog(sqlType, tableName, beforeImage, afterImage);
		connectionContext.addUndoLog(undoLog);
	}

	private void recognizeLockKeyConflictException(GlobalTransactionException e) throws SQLException
	{
		if (e.getTransactionExceptionCode() == TransactionExceptionCode.lock_conflict)
		{
			throw new LockConflictException();
		} else
		{
			if (Constants.developMode)
			{
				e.printStackTrace();
			}
			log.error($.printStackTraceToString(e));
			//
			throw new SQLException(e);
		}
	}

	private void report(boolean committed) throws SQLException
	{
		int retry = 5; // TODO: configure
		while (retry > 0)
		{
			try
			{
				ResourceManager.get().branchTransactionReport(connectionContext.getGlobalTransactionIdWithAddressAndPort(), connectionContext.getBranchTransactionId(), (committed ? BranchTransactionStatus.phase_one_succeed : BranchTransactionStatus.phase_one_failed), null);
				return;
			} catch (Throwable ex)
			{
				log.error("failed to report [" + connectionContext.getBranchTransactionId() + "/" + connectionContext.getGlobalTransactionIdWithAddressAndPort() + "] globalTransactionCommit done [" + committed + "] Retry Countdown: " + retry);
				retry--;

				if (retry == 0)
				{
					throw new SQLException("failed to report branch branchTransactionStatus " + committed, ex);
				}
			}
		}
	}

	@Override
	public void rollback() throws SQLException
	{
		connection.rollback();
		if (connectionContext.inGlobalTransaction())
		{
			if (connectionContext.isBranchRegistered())
			{
				report(false);
			}
		}
		connectionContext.reset();// ?
	}

	//
	@Override
	public void setAutoCommit(boolean autoCommit) throws SQLException
	{
		if ((autoCommit) && !getAutoCommit())
		{
			// change autocommit from false to true, we should globalTransactionCommit()
			// first according to JDBC spec.
			commit();
		}
		connection.setAutoCommit(autoCommit);
	}
}
